import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;
import java.sql.*;

public class queue {
    public static BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
    public static Queue<Message> SaveQueue = new ArrayDeque<>();
    public static Semaphore semaphore = new Semaphore(0);
    public static Queue<Socket> consumerQueue = new ArrayDeque<>();//用于存储注册的消费者
    // 注册消费者
    public static void registerConsumer() {
        try {
            semaphore.acquire();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    static final String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
    static final String DB_URL = "jdbc:mysql://localhost:3306/分布式实验?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC";

    static final String USER = "root";
    static final String PASS = "123456789";

    public static void readQueue() throws SQLException, InterruptedException {
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/textual?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC", "root", "123456789");
        String sql;
        sql = "SELECT id,timestamp,tag,body FROM fbdata";
        PreparedStatement preparedStatement2 = connection.prepareStatement(sql);
        ResultSet rs = preparedStatement2.executeQuery();

        // 12 小时对应的秒数
        int twelveHoursInSeconds = 12 * 60 * 60; // 43200秒
        
        int timestamp = 0;
        String tag = "";
        String body = "";
        int id=0;
        while (rs.next()) {
            //读取数据
            id = rs.getInt("id");
            timestamp = rs.getInt("timestamp");
            tag = rs.getString("tag");
            body = rs.getString("body");
            System.out.println(id+" "+timestamp+" "+tag+" "+body);
            //将数据存入队列中
            LocalDateTime currentDateTime = LocalDateTime.now();
            int timeStampInSeconds = (int) currentDateTime.toEpochSecond(ZoneOffset.UTC);
            if (Math.abs(timeStampInSeconds - timestamp)>=twelveHoursInSeconds) {
                System.out.println("超时信息！");
            }else {
                Message msg = new Message(tag,body);
                msg.setId(id);
                msg.setTimestamp(timestamp);
                queue.put(msg);
                semaphore.release();
            }
        }

        rs.close();
        preparedStatement2.close();
    }

    public static void saveMessage(Message message) throws ClassNotFoundException, SQLException {
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/textual?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC", "root", "123456789");
        String sql2;
        System.out.println("SAVING MESSAGE");
        sql2 = "INSERT INTO fbdata(id,timestamp,tag,body) VALUES(?,?,?,?)";
        PreparedStatement preparedStatement = connection.prepareStatement(sql2);
        preparedStatement.setInt(1,message.getId());
        preparedStatement.setInt(2,message.getTimestamp());
        preparedStatement.setString(3,message.getTag());
        preparedStatement.setString(4,message.getBody());
        int n = preparedStatement.executeUpdate();
        preparedStatement.close();

    }
        // 生产者发送消息
    public static void sendMessage() {
        semaphore.release();
    }

    // 消息队列的消息推送逻辑（可以改进为消息持久化等）
    public Message getMessage() throws InterruptedException {
        return queue.take(); // 阻塞等待消息
    }

    public static void main(String[] args) throws IOException, InterruptedException, SQLException, ClassNotFoundException {
        int port1 = 8888;
        int port2 = 8886;

        Message msg;
        SendMsg sendMsg = new SendMsg();
        readQueue();
        ServerSocket serverSocket1 = new ServerSocket(port1);//producer
        System.out.println("Listening on port " + port1);
        ServerSocket serverSocket2 = new ServerSocket(port2);//consumer
        System.out.println("Listening on port " + port2);
        ExecutorService socketThread1 = Executors.newFixedThreadPool(2);
        ExecutorService socketThread2 = Executors.newFixedThreadPool(2);
        ExecutorService registerThreadPool = Executors.newFixedThreadPool(2);

        Runnable runnable1 = ()-> {
            Message message;
            try{
                byte[] b = new byte[1024];
                Socket socket1 = serverSocket1.accept();
                InputStream in = socket1.getInputStream();
                OutputStream out = socket1.getOutputStream();
                ObjectInputStream ois = new ObjectInputStream(in);
                message = (Message) ois.readObject();
                queue.put(message);
                sendMessage();
                SaveQueue.add(message);
                if (SaveQueue.size()>100){
                    while (SaveQueue.size()>0){
                        Message saveMessage = SaveQueue.remove();
                        saveMessage(saveMessage);
                    }
                }
                System.out.println("Message received: " + message.getBody());
                out.write(message.getBody().getBytes());
                ois.close();
                in.close();
                out.close();
                socket1.close();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        };
        Runnable runnable2 = ()-> {
            try{
                byte[] b = new byte[1024];
                Socket socket2 = consumerQueue.poll();
                OutputStream out = socket2.getOutputStream();
                ObjectOutputStream oos = new ObjectOutputStream(out);
                InputStream inputStream = socket2.getInputStream();

                //获取消费者的消费模式
                String opt = "";
                StringBuilder sb = new StringBuilder();
                int len;
                while ((len = inputStream.read(b)) != -1) {
                    sb.append(new String(b, 0, len));
                }

                opt = sb.toString();
                System.out.println(opt);
                Message message;//消息

                if (opt.equals("rent")){
                    if (sendMsg.getNum()==0){
                        registerConsumer();
                    }
                    if (sendMsg.getMsg()==null){
                        sendMsg.setMsg(queue.take());
                    }
                    else if (sendMsg.getNum()>=2){
                        sendMsg.setMsg(queue.take());
                        sendMsg.reSetNum();
                    }
                    message = sendMsg.getMsg();
                    oos.writeObject(message);
                    sendMsg.addNum();
                    System.out.println("广播模式发送");
                }else{
                    if(!queue.isEmpty()){
                        registerConsumer();
                        message = queue.take();
                        oos.writeObject(message);
                    }
                }


//                message = queue.take();
//                oos.writeObject(message);
//                out.write(message.getBody().getBytes());
                oos.close();
                out.close();
                socket2.close();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        };

        Runnable runnable3 = ()-> {
            try{
                byte[] b = new byte[1024];
                Socket consumerSocket = serverSocket2.accept();
                consumerQueue.add(consumerSocket);
                System.out.println("Consumer received: " + consumerQueue.size());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }

        };

        while (true) {
            registerThreadPool.submit(runnable3);//多线程注册
            socketThread2.submit(runnable2);//多线程监听生产者
            socketThread1.submit(runnable1);//多线程广播消费者
        }

    }
}

class SendMsg{
    Message msg = null;
    int num = 0;

    public void setMsg(Message msg) {
        this.msg = msg;
    }

    public void addNum(){
        num++;
    }

    public int getNum(){
        return num;
    }

    public Message getMsg() {
        return msg;
    }

    public void setNum(int num){
        this.num = num;
    }
    public void reSetNum(){
        if(num >= 2){
            num = 0;
        }
    }
}